package com.example.common.common.config.bloom;

import lombok.RequiredArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;

@Configuration
public class BloomFilterConfig {

    private static final Logger log = LoggerFactory.getLogger(BloomFilterConfig.class);
    private static final String BLOOM_FILTER_EXISTS_KEY = "bf.exists";

    @Value("${spring.data.redis.bloom.key:bloom:default}")
    private String bloomKey;

    @Value("${spring.data.redis.bloom.rate:0.01}")
    private double errorRate;

    @Value("${spring.data.redis.bloom.capacity:10000}")
    private long capacity;

    /**
     * RedisTemplate 配置
     */
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(connectionFactory);
        template.setKeySerializer(new StringRedisSerializer());
        template.setHashKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new StringRedisSerializer());
        template.setHashValueSerializer(new StringRedisSerializer());
        template.afterPropertiesSet();
        return template;
    }

    /**
     * 布隆过滤器服务
     */
    @Bean
    public BloomFilterService bloomFilterService(RedisTemplate<String, Object> redisTemplate) {
        return new BloomFilterService(redisTemplate);
    }

    /**
     * 布隆过滤器包装 Bean（固定 key）
     */
    @Bean(name = "shortLinkBloomFilter")
    public BloomFilterWrapper shortLinkBloomFilter(BloomFilterService bloomFilterService) {
        // 自动初始化
        bloomFilterService.initFilterIfNeeded(bloomKey, errorRate, capacity);
        return new BloomFilterWrapper(bloomFilterService, bloomKey);
    }

    @RequiredArgsConstructor
    public static class BloomFilterService {
        private final RedisTemplate<String, Object> redisTemplate;
        private static final String RESERVE_COMMAND = "BF.RESERVE";
        private static final String ADD_COMMAND = "BF.ADD";
        private static final String EXISTS_COMMAND = "BF.EXISTS";

        public boolean filterExists(String key) {
            return Boolean.TRUE.equals(redisTemplate.hasKey(key));
        }

        public void initFilterIfNeeded(String key, double errorRate, long capacity) {
            if (!filterExists(key)) {
                try {
                    initFilter(key, errorRate, capacity);
                    log.info("Bloom filter initialized: key={}, errorRate={}, capacity={}", key, errorRate, capacity);
                } catch (Exception e) {
                    log.error("Failed to initialize bloom filter: {}", key, e);
                }
            }
        }

        public void initFilter(String key, double errorRate, long capacity) {
            redisTemplate.execute((RedisConnection connection) -> {
                Object result = connection.execute(RESERVE_COMMAND, new byte[][]{
                        key.getBytes(StandardCharsets.UTF_8),
                        doubleToBytes(errorRate),
                        longToBytes(capacity)
                });

                // 检查操作是否成功
                if (result != null && !"OK".equals(result.toString())) {
                    throw new BloomFilterException("Bloom filter init failed: " + result);
                }
                return null;
            }, true);
        }

        public boolean add(String key, String value) {
            return executeBooleanCommand(ADD_COMMAND, key, value);
        }

        public boolean exists(String key, String value) {
            return executeBooleanCommand(EXISTS_COMMAND, key, value);
        }

        private boolean executeBooleanCommand(String command, String key, String value) {
            Object result = redisTemplate.execute((RedisConnection connection) ->
                    connection.execute(command, new byte[][]{
                            key.getBytes(StandardCharsets.UTF_8),
                            value.getBytes(StandardCharsets.UTF_8)
                    }), true);

            return parseBooleanResult(result);
        }

        private boolean parseBooleanResult(Object result) {
            if (result instanceof Long) {
                return (Long) result == 1L;
            } else if (result instanceof Integer) {
                return (Integer) result == 1;
            } else if (result instanceof Boolean) {
                return (Boolean) result;
            }
            return false;
        }

        private byte[] longToBytes(long value) {
            return Long.toString(value).getBytes(StandardCharsets.UTF_8);
        }

        private byte[] doubleToBytes(double value) {
            // 使用BigDecimal避免科学计数法
            return new BigDecimal(Double.toString(value)).toPlainString().getBytes(StandardCharsets.UTF_8);
        }
    }

    @RequiredArgsConstructor
    public static class BloomFilterWrapper {
        private final BloomFilterService delegate;
        private final String key;

        public boolean add(String value) {
            return delegate.add(key, value);
        }

        public boolean exists(String value) {
            return delegate.exists(key, value);
        }

        public boolean filterExists() {
            return delegate.filterExists(key);
        }

        public void reinitialize(double errorRate, long capacity) {
            delegate.initFilter(key, errorRate, capacity);
        }
    }

    public static class BloomFilterException extends RuntimeException {
        public BloomFilterException(String message) {
            super(message);
        }
    }
}